package com.billdimmick.merkabah;

import static org.junit.Assert.*;

import java.security.PrivateKey;
import java.security.PublicKey;
import java.security.cert.Certificate;
import java.util.Collection;
import java.util.Properties;
import java.util.UUID;

import org.apache.commons.lang.Validate;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class TestIntegSQSManager {

	private static final Log log=LogFactory.getLog(TestIntegSQSManager.class);
	private static final long MESSAGE_CONSISTENCY_TIMEOUT=500;
	
	private SQSManager manager;
		
	@Before
	public void setUp() throws Exception {
		Properties properties=new Properties();
		properties.load(ClassLoader.getSystemResourceAsStream("integ.properties"));
		SQSSecurityAdapter adapter=new BasicSQSSecurityAdapter(properties);
		manager=new DelaySQSManager(adapter,MESSAGE_CONSISTENCY_TIMEOUT);
	}

	@After
	public void tearDown() throws Exception {
	}

	@Test
	public final void testFailedAdapterConstructor() {
		constructorTestExpectFailure(null);
		constructorTestExpectFailure(new IntegSecurityAdapter(null, UUID.randomUUID().toString()));
		constructorTestExpectFailure(new IntegSecurityAdapter("", UUID.randomUUID().toString()));
		constructorTestExpectFailure(new IntegSecurityAdapter(UUID.randomUUID().toString(), null));
		constructorTestExpectFailure(new IntegSecurityAdapter(UUID.randomUUID().toString(), ""));
	}
	
	@Test
	public final void testStringMessageRoundtrip() throws Exception {
		final String queue=UUID.randomUUID().toString();
		final String body=UUID.randomUUID().toString();
		createQueue(queue);
		manager.send(queue, body);
		Message message=manager.receive(queue);
		assertNotNull(message);
		assertNotNull(message.getMessageId());
		assertNotNull(message.getReceiptHandle());
		assertNotNull(message.getQueue());
		assertNotNull(message.getMessage());
		assertEquals(body, message.getMessage());
		manager.delete(message);
		deleteQueue(queue);		
	}

	@Test
	public final void testObjectMessageRoundtrip() throws Exception {
		final String queue=UUID.randomUUID().toString();
		final String body=UUID.randomUUID().toString();
		createQueue(queue);
		
		Message message=new Message(queue, body);
		assertNotNull(message.getQueue());
		assertNotNull(message.getMessage());
		assertNull(message.getMessageId());
		assertNull(message.getReceiptHandle());
		
		manager.send(message);
		assertNotNull(message.getQueue());
		assertNotNull(message.getMessage());
		assertNotNull(message.getMessageId());
		assertNull(message.getReceiptHandle());
		
		Message receipt=manager.receive(queue);
		assertNotNull(receipt);
		assertNotNull(receipt.getMessageId());
		assertNotNull(receipt.getReceiptHandle());
		assertNotNull(receipt.getQueue());
		assertNotNull(receipt.getMessage());
		assertEquals(message.getQueue(), receipt.getQueue());
		assertEquals(message.getMessageId(), receipt.getMessageId());
		assertEquals(message.getMessage(), receipt.getMessage());
		
		manager.delete(receipt);
		deleteQueue(queue);		
	}

	
	@Test
	public final void testCreateQueue() throws Exception {
		Collection<String> baseline=manager.listQueues();
		log.info(String.format("Baseline := %s", baseline.toString()));
		String name=UUID.randomUUID().toString();
		log.info(String.format("Creating '%s'", name));
		createQueue(name);
		Collection<String> afterCreate=manager.listQueues();
		log.info(String.format("After Create := %s", afterCreate.toString()));		
		assertTrue(afterCreate.contains(name));
		assertFalse(baseline.contains(name));
		deleteQueue(name);
	}

	@Test
	public final void testDeleteQueue() throws Exception {
		Collection<String> baseline=manager.listQueues();
		log.info(String.format("Baseline := %s", baseline.toString()));
		String name=UUID.randomUUID().toString();
		log.info(String.format("Creating '%s'", name));
		createQueue(name);
		Collection<String> afterCreate=manager.listQueues();
		assertFalse(baseline.contains(name));
		assertTrue(afterCreate.contains(name));
		log.info(String.format("After Create := %s", afterCreate.toString()));		
		deleteQueue(name);
		Collection<String> afterDelete=manager.listQueues();
		assertFalse(afterDelete.contains(name));
	}
		
	private void constructorTestExpectFailure(final SQSSecurityAdapter adapter) {
		try {
			new SQSManager(adapter);
			fail("Expected exception on null-parameter constructor.");
		} catch (IllegalArgumentException iae) {
		}
	}
	
	/**
	 * Creation of queues has a massive consistency timeout - integration tests will 
	 * need to sleep or we'll get false negatives...  
	 */
	private void createQueue(final String name) throws SQSException {		
		this.manager.createQueue(name);
		try {
			Thread.sleep(20000);
		} catch (final InterruptedException ie) {}
	}

	/**
	 * Deletion of queues has a massive consistency timeout - integration tests will 
	 * need to sleep or we'll get false negatives...  
	 */
	private void deleteQueue(final String name) throws SQSException {
		this.manager.deleteQueue(name);
		try {
			Thread.sleep(20000);
		} catch (final InterruptedException ie) {}
	}

	
	public static class IntegSecurityAdapter extends SQSSecurityAdapter {		
		private final String id;
		private final String key;
		
		public IntegSecurityAdapter(final String id, final String key) {
			this.id=id;
			this.key=key;
		}
		
		public String getAWSAccessId() {
			return this.id;
		}

		public String getAWSSecretKey() {
			return this.key;
		}
		
		protected Certificate getCertificate(String name) { return null; }
		protected PrivateKey getPrivateKey(String name) { return null; }
		protected PublicKey getPublicKey(String name) { return null; }
		public String getDefaultEncryptionKey() { return "default-crypt"; }
		public String getDefaultSigningKey() { return "default-sign"; }
		
	}
	
	public static class DelaySQSManager extends SQSManager {

		private final static Log log=LogFactory.getLog(DelaySQSManager.class);
		
		private final long messageDelay;
		
		public DelaySQSManager(final SQSSecurityAdapter credentials, final long messageDelay) {
			super(credentials);
			Validate.isTrue(messageDelay>0, "Delay must be greater than zero.");
			this.messageDelay=messageDelay;
		}

		@Override
		public void createQueue(String name, int timeout) throws SQSException {
			super.createQueue(name, timeout);		
		}
		
		@Override
		public void delete(Message message) throws SQSException {
			sleepForMessage();
			super.delete(message);
		}
		
		@Override
		public void deleteQueue(String name) throws SQSException {
			super.deleteQueue(name);
		}
		
		@Override
		public int getApproximateNumberOfMessages(String name) throws SQSException {
			return super.getApproximateNumberOfMessages(name);
		}
		
		@Override
		public Collection<String> listQueues(String prefix) throws SQSException {
			return super.listQueues(prefix);
		}
		
		@Override
		public Collection<Message> receive(String queue, int max, int timeout)
				throws SQSException {
			sleepForMessage();
			return super.receive(queue, max, timeout);
		}
		
		@Override
		public void send(Message message) throws SQSException {
			sleepForMessage();
			super.send(message);
		}
		
		private void sleepForMessage() {
			try {
				Thread.sleep(this.messageDelay);
			} catch (InterruptedException ie) {
				log.error("Interrupted?  This *should* be ok...", ie);
			}
		}

	}
}
